package cas.ihep.test;

import cas.ihep.hdfs.HdfsBlock;
import cas.ihep.hdfs.HdfsFile;
import cas.ihep.hdfs.HdfsSystem;
import com.google.common.io.Closer;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.JobID;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.File;
import java.io.IOException;

public class ExportMapper extends Mapper<JobID, Path, IntWritable,IntWritable> {

    public void run(Context context) throws IOException, InterruptedException {
        setup(context);
        JobID jobId = context.getJobID();
        FileSplit split = (FileSplit)context.getInputSplit();
        Path path = split.getPath();
        map(jobId, path, context);
        cleanup(context);
    }
    protected void map(JobID id,Path path,Context context)throws IOException,InterruptedException{
        try (Closer closer= Closer.create()){
            HdfsSystem hdfs = closer.register(new HdfsSystem(context.getConfiguration().get("fs.defaultFS"), context.getConfiguration()));
            HdfsFile file=closer.register(hdfs.open(path.getName()));
            HdfsBlock blk=closer.register(file.getBlock(0));
            String blkpath=blk.getBlockPath();
            ProcessBuilder builder=new ProcessBuilder();
            builder.command("globus-url-copy",blkpath,"ftp://hadoop:123456@helion01.ihep.ac.cn:5555/"+context.getConfiguration().get("target.directory"));
            builder.redirectError(new File("/var/log/hadoop/job/"+id.toString()+".err"));
            builder.redirectOutput(new File("/var/log/hadoop/job/"+id.toString()+".out"));
            Process process=builder.start();
            int exitValue=process.waitFor();
            if (exitValue==0){
                return ;
            }
            System.err.println("Error");
        }catch(Exception e){
            e.printStackTrace();
            throw new IOException(e);
        }
    }
}
